package com.imooc.web;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;

public class DistributeLock {
	private CuratorFramework client = null;

	final static Logger log = LoggerFactory.getLogger(DistributeLock.class);

	private static CountDownLatch zkLockLatch = new CountDownLatch(1);

	private static final String ZK_LOCK_PROJECT = "imooc-locks";

	private static final String DISTRIBUTED_LOCK = "distributed_lock";

	public DistributeLock(CuratorFramework client) {
		this.client = client;
	}

	public void init(){
		client = client.usingNamespace("ZKLocks-Namespace");

		try {
			if (client.checkExists().forPath("/" + ZK_LOCK_PROJECT) == null) {
				client.create().creatingParentsIfNeeded()
						.withMode(CreateMode.PERSISTENT)
						.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
						.forPath("/"+ZK_LOCK_PROJECT);
			}
			//针对zk的分布式锁节点,创建相应的watcher事件监听
			addWatcherToLock("/"+ZK_LOCK_PROJECT);

		} catch (Exception e) {
			log.error("客户端链接zookeeper服务器错误...请重试...");
		}
	}


	public void getLock(){
		//使用死循环,当且仅当上一个锁释放并且当前请求获得锁成功才会跳出
		while (true) {
			try {
				client.create()
						.creatingParentsIfNeeded()
						.withMode(CreateMode.EPHEMERAL)
						.withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
						.forPath("/" + ZK_LOCK_PROJECT + "/" + DISTRIBUTED_LOCK);
				log.info("获得分布式锁成功");
				return;//如果锁的节点能被创建成功,则锁没有被占用
			} catch (Exception e) {
				log.info("获得分布式锁失败...");
				try {
					//如果没有获取到锁,需要重新设置同步资源值
					if (zkLockLatch.getCount() <= 0) {
						zkLockLatch = new CountDownLatch(1);
					}
					//阻塞线程
					zkLockLatch.await();
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}

			}
		}
	}

	public void addWatcherToLock(String path) throws Exception {
		final PathChildrenCache cache = new PathChildrenCache(client, path, true);
		cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
		cache.getListenable().addListener((client,event)->{
			if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
				String paths = event.getData().getPath();
				log.info("上一个会话已释放锁或改会话已断开,节点路径为:" + path);
				if (path.contains(DISTRIBUTED_LOCK)) {
					log.info("释放计数器,让当前请求来获得分布式锁");
					zkLockLatch.countDown();
				}
			}
		});
	}

	public boolean releaseLock(){
		try {
			if (client.checkExists().forPath("/" + ZK_LOCK_PROJECT + "/" + DISTRIBUTED_LOCK) != null) {
				client.delete().forPath("/" + ZK_LOCK_PROJECT + "/" + DISTRIBUTED_LOCK);
			}
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
		log.info("分布式说释放完毕");
		return true;
	}
}
